package top.ninwoo.cloudcenter.service.impl;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;
import top.ninwoo.cloudcenter.register.CloudRegisterCenter;
import top.ninwoo.cloudcenter.service.CloudService;
import top.ninwoo.common.entity.*;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.*;

@Service
public class CloudServiceImpl implements CloudService {
    private static final Logger LOG = LoggerFactory.getLogger(CloudServiceImpl.class);
    private static final Random randomInt = new Random(14);
    private static final String CREATE_CLUSTER = "/createCluster";
    private static final String DELETE_CLUSTER = "/delCluster";
    private static final String ADJUST_CLUSTER = "/adjustCluster";
    private static final String SEND_LOGICTOPO = "/sendLogicTopo";
    private static final String ADJUST_LOGICTOPO = "/adjustLogicTopo";
    private static final String ADJUST_LOGICTOPO_BYTOPO = "/adjustLogicTopoBytopo";
    private static final String DROP_DOCKER_NETWORK = "/dropDockerNetwork?clusterId={clusterId}&appName={appName}&ipList={ipList}";
    private static final String CANCEL_DROP_DOCKER_NETWORK = "/cancelDropDockerNetwork?clusterId={clusterId}&appName={appName}&ipList={ipList}";
    private static final String REMOTE_IP_LIST_BY_APPNAME = "/getIpListByAppName?clusterId={clusterId}&appName={appName}";
    private static final String ENABLE_NETWORK_MONITOR = "/enableContainerMonitor?clusterId={clusterId}&containerName={appName}";
    private static final String CANCEL_NETWORK_MONITOR = "/cancelNetworkMonitor?clusterId={clusterId}&containerName={appName}";
    private static final String ADD_QOS = "/addQos?clusterId={clusterId}&cName={cName}&maxRate={maxRate}&latency={latency}";
    private static final ExecutorService initEdgeExecutorPool = Executors.newFixedThreadPool(6 * 2);
    private static final CompletionService<ThreadResult> initEdgeExcutorCompletionPool = new ExecutorCompletionService<>(initEdgeExecutorPool);

    @Value("${bs.ipservice.url}")
    private String ipServiceUrl;
    // 全部的逻辑拓扑
    // 全部的逻辑拓扑
    private ConcurrentHashMap<Long, NetworkTopology> allLogicalTopo = new ConcurrentHashMap<>();
    // 用于存储配置文件
    private ConcurrentHashMap<Long, List<SeparatedClusterConfig>> allClusterConfig
            = new ConcurrentHashMap<>();
    // 存储app在哪一个边缘节点
    private ConcurrentHashMap<Long, Map<String, Set<String>>> allAppStatus
            = new ConcurrentHashMap<>();

    private ConcurrentHashMap<Long, Map<String, Map<String, Set<String>>>> allContainerIds = new ConcurrentHashMap<>();
    @Resource
    private CloudRegisterCenter cloudRegisterCenter;

    @Resource
    private RestTemplate restTemplate;

    @Override
    public Boolean addQos(Long clusterId, String appName, String maxRate, String latency) {
        Map<String, Map<String, Set<String>>> stringMapMap = allContainerIds.get(clusterId);
        String nodeId = null;
        for (String s : stringMapMap.keySet()) {
            if (stringMapMap.get(s).keySet().contains(appName)) {
                nodeId = s;
            }
        }
        if(nodeId == null) {
            throw new RuntimeException("无法找到可以使用的节点");
        }
        HashMap<String, Object> param = new HashMap<>();
        param.put("clusterId", clusterId);
        param.put("cName", appName);
        param.put("maxRate", maxRate);
        param.put("latency", latency);
        Boolean res = restTemplate.getForObject("http://" + nodeId + ADD_QOS, Boolean.class, param);
        return res;
    }

    @Override
    public boolean adjustCluster(List<SeparatedClusterConfig> clusterConfigs) {
        if(clusterConfigs == null) {
            throw new RuntimeException("clusterConfig cannot be null.");
        }
        try {
            clusterConfigs.forEach(c -> {
                ResponseEntity<String> response = restTemplate.postForEntity("http://" + c.getEdgeNodeId() + ADJUST_CLUSTER, c.getClusterConfig(), String.class);
                if (!response.getStatusCode().is2xxSuccessful()) {
                    throw new RuntimeException("send Error!");
                }
                LOG.info("{} update success!", c.getEdgeNodeId());
            });
        } catch (Exception e) {
            return false;
        }

        return true;
    }

    @Override
    public boolean sendLogicTopo(List<SeparatedClusterConfig> clusterConfigs) {
        if(clusterConfigs == null) {
            throw new RuntimeException("clusterConfig cannot be null.");
        }
        try {
            clusterConfigs.forEach(c -> {
                ResponseEntity<String> response = restTemplate.postForEntity("http://" + c.getEdgeNodeId() + SEND_LOGICTOPO, c.getClusterConfig(), String.class);
                if (!response.getStatusCode().is2xxSuccessful()) {
                    throw new RuntimeException("send Error!");
                }
                LOG.info("{} send logictopo success!", c.getEdgeNodeId());
            });
        } catch (Exception e) {
            return false;
        }

        return true;
    }

    @Override
    public boolean adjustLogicTopo(List<SeparatedClusterConfig> clusterConfigs) {
        if(clusterConfigs == null) {
            throw new RuntimeException("clusterConfig cannot be null.");
        }
        try {
            clusterConfigs.forEach(c -> {
                ResponseEntity<String> response = restTemplate.postForEntity("http://" + c.getEdgeNodeId() + ADJUST_LOGICTOPO, c.getClusterConfig(), String.class);
                if (!response.getStatusCode().is2xxSuccessful()) {
                    throw new RuntimeException("send Error!");
                }
                LOG.info("{} adjust logictopo success!", c.getEdgeNodeId());
            });
        } catch (Exception e) {
            return false;
        }

        return true;
    }

    /**
     * 这里的拓扑必须限定为逻辑拓扑
     * @param clusterConfig
     * @return
     */
    @Override
    public String initCluster(ClusterConfig clusterConfig) {

        // TODO: step1: 校验配置是否合法
        check(clusterConfig);

        // 在数据库中注册该节点
        registerClusterInDataBase(clusterConfig);
        // TODO: step2: 拆分配置

        // TODO: step3: 下发配置到不同的边缘服务节点
        // TODO: 保存配置和边缘节点的映射关系
        return "";
    }

    @Override
    public Set<String> getIpListFromRemoteIpServiceByAppName(long clusterId, String appName) {
        Map<String, Object> param = new HashMap<>();
        param.put("clusterId", clusterId);
        param.put("appName", appName);
        ResponseEntity<Set> restEntity = restTemplate.getForEntity("http://" + ipServiceUrl + REMOTE_IP_LIST_BY_APPNAME, Set.class, param);
        if(!restEntity.getStatusCode().is2xxSuccessful()) {
            LOG.error("远程IP服务无法连接");
            throw new RuntimeException("远程服务无法连接：" + ipServiceUrl + REMOTE_IP_LIST_BY_APPNAME);
        }
        return restEntity.getBody();
    }

    private ClusterConfig registerClusterInDataBase(ClusterConfig clusterConfig) {
        clusterConfig.setId(11111L);
        return clusterConfig;
    }

    private boolean check(ClusterConfig clusterConfig) {
        return true;
    }

    private List<SeparatedClusterConfig> separateClusterConfig(ClusterConfig clusterConfig) {
        // 先校验clusterConfig的模式
        String type = clusterConfig.getType();
        if("single".equals(type)) {
            return singleSeparateClusterConfig(clusterConfig);
        }
        if("random".equals(type)) {
            return randomSeparateClusterConfig(clusterConfig);
        }
        if("all".equals(type)) {
            return allSeparatedClusterConfig(clusterConfig);
        }
        if(type.startsWith("size:")) {
            int size = Integer.parseInt(type.split(":")[1]);
            return fixedSpeparatedClusterConfig(clusterConfig, size);
        }
        throw new IllegalArgumentException("clusterConfig's " + type + " is Illegal!");
    }

    private List<SeparatedClusterConfig> fixedSpeparatedClusterConfig(ClusterConfig clusterConfig, int size) {

        List<String> nodes = chooseEdgeNodes(size);

        Map<String, SeparatedClusterConfig> separatedClusterConfigs = new HashMap<>();

        for (int i = 0; i < nodes.size(); i++) {
            SeparatedClusterConfig separatedClusterConfig = new SeparatedClusterConfig();
            separatedClusterConfig.setEdgeNodeId(nodes.get(i));
            ClusterConfig config = new ClusterConfig();
            config.setId(clusterConfig.getId());
            // 只有这里不同
            config.setDockers(new ArrayList<>());
            NetworkTopology networkTopology = new NetworkTopology();
            config.setTopology(networkTopology);
            config.setOwner(clusterConfig.getOwner());
            config.setCreateTime(clusterConfig.getCreateTime());
            config.setType(clusterConfig.getType());
            separatedClusterConfig.setClusterConfig(config);
            separatedClusterConfigs.put(nodes.get(i), separatedClusterConfig);

        }


        int index = 0;
        // TODO: 拆分配置,构建出新的配置
        List<ContainerDescription> dockers = clusterConfig.getDockers();
        for (ContainerDescription docker : dockers) {
            if("one".equals(docker.getMode())) {
                // 该docker全部部署到同一个集群上
                // TODO: 这里可以添加一些选择标准
                SeparatedClusterConfig separatedClusterConfig = separatedClusterConfigs.get(nodes.get(index));
                ClusterConfig edgeClusterConfig = separatedClusterConfig.getClusterConfig();
                // 更新edgeColusterClusterConfig上的配置
                edgeClusterConfig.getDockers().add(docker);
                NetworkTopology topology = edgeClusterConfig.getTopology();
                String[] appNames = topology.getAppNames();
                int[][] topo = topology.getTopology();
                // TODO: 扩大规模
                if(appNames == null) {
                    appNames = new String[]{docker.getDockerContainer().getName()};
                    topo = new int[][]{{0}};
                } else {
                    appNames = Arrays.copyOf(appNames, appNames.length + 1);
                    appNames[appNames.length - 1] = docker.getDockerContainer().getName();
                    // 可能出问题
                    topo = new int[topo.length + 1][topo.length + 1];
                }
                topology.setAppNames(appNames);
                topology.setTopology(topo);
            }
            else if(docker.getMode().equals("roundbin")) {
                int replicas = docker.getReplicas();
                int average = replicas / nodes.size();
                for (int i = 0; i < nodes.size(); i++) {
                    // TODO: 更新集群，更新
                }
            }

            index = (index + 1) % nodes.size();
        }
        return null;
    }

    private List<SeparatedClusterConfig> allSeparatedClusterConfig(ClusterConfig clusterConfig) {
        List<String> availableEdgeNodes = cloudRegisterCenter.getAvailableEdgeNodes();
        return null;
    }

    private List<SeparatedClusterConfig> randomSeparateClusterConfig(ClusterConfig clusterConfig) {
        // 创建随机的
        return null;
    }

    private List<SeparatedClusterConfig> singleSeparateClusterConfig(ClusterConfig clusterConfig) {
        // 选择一个节点
        List<String> nodes = chooseEdgeNodes(1);
        return null;
    }

    private List<String> chooseEdgeNodes(int num) {
        // TODO 可以扩展更多选择方案，比如已计算优先，或者内存优先

        ArrayList<String> res = new ArrayList<>(num);

        // step1: 获取全部可用的边缘节点
        List<String> availableEdgeNodes = cloudRegisterCenter.getAvailableEdgeNodes();

        if(availableEdgeNodes.size() <= num) {
            return availableEdgeNodes;
        }

        // step2: 随机挑选一个节点
        for (int i = 0; i < num; i++) {
            int index = randomInt.nextInt(availableEdgeNodes.size());
            res.add(availableEdgeNodes.get(index));
            availableEdgeNodes.remove(index);
        }
        return res;
    }

    /**
     * 下发clusterConfigs到边缘节点
     * @param clusterConfigs
     * @return
     */
    @Override
    public List<SeparatedClusterConfig>  sendClusterConfigToEdgeNode(List<SeparatedClusterConfig> clusterConfigs) {
        if(clusterConfigs.size() == 0) {
            LOG.warn("下发的集群配置为空。");
            return clusterConfigs;
        }
        List<String> allAppNames = new ArrayList<>();
        // 这里应该做一个判断，edgeNode的ip是否可用，如果不可用，或者未设定，将随机挑选一个进行设置
        // TODO:这里使用一个线程进行创建，加快docker启动速度
        clusterConfigs.forEach(c -> {
            if (c.getEdgeNodeId() == null || "".equals(c.getEdgeNodeId())) {
                String edgeNode = chooseEdgeNodes(1).get(0);
                c.setEdgeNodeId(edgeNode);
            }
            // 下发配置
            ResponseEntity<Map> response = restTemplate.postForEntity("http://" + c.getEdgeNodeId() + CREATE_CLUSTER, c.getClusterConfig(), Map.class);
            if(!response.getStatusCode().is2xxSuccessful()) {
                LOG.error("集群配置下发失败:{}:{}", c.getEdgeNodeId(), response.getBody());

            }
            // 更新配置
            //c.setClusterConfig(response.getBody());
            // TODO: 存储appName和containerId的映射
            if (!allContainerIds.containsKey(c.getClusterConfig().getId())) {
                allContainerIds.put(c.getClusterConfig().getId(), new HashMap<>());
            }
            Map<String, Map<String, Set<String>>> allNames = allContainerIds.get(c.getClusterConfig().getId());
            if (!allNames.containsKey(c.getEdgeNodeId())) {
                allNames.put(c.getEdgeNodeId(), new HashMap<>());
            }
            Map<String, Set<String>> edgeContainerIds = allNames.get(c.getEdgeNodeId());
            Map<String, List<String>> nameToIds = response.getBody();
            nameToIds.forEach((name, v) -> {
                if(!edgeContainerIds.containsKey(name)) {
                    edgeContainerIds.put(name, new HashSet<>());
                }
                edgeContainerIds.get(name).addAll(v);
            });
            // TODO: 对应需要设计一个删除这个配置的操作
            // 构建初始化的逻辑拓扑
            for (String appName : c.getClusterConfig().getTopology().getAppNames()) {
                if(!appName.startsWith("br")) {
                    allAppNames.add(appName);
                    if (!allAppStatus.containsKey(c.getClusterConfig().getId())) {
                        allAppStatus.put(c.getClusterConfig().getId(), new HashMap<>());
                    }
                    // 保存appnames
                    Map<String, Set<String>> clusterAppConfig = allAppStatus.get(c.getClusterConfig().getId());
                    if(!clusterAppConfig.containsKey(appName)) {
                        clusterAppConfig.put(appName, new HashSet<>());
                    }
                    Set<String> edgeNodes = clusterAppConfig.get(appName);
                    edgeNodes.add(c.getEdgeNodeId());
                }
            }
            if(!allClusterConfig.containsKey(c.getClusterConfig().getId())) {
                allClusterConfig.put(c.getClusterConfig().getId(), new ArrayList<>());
            }
            allClusterConfig.get(c.getClusterConfig().getId()).add(c);
        });

        allLogicalTopo.put(clusterConfigs.get(0).getClusterConfig().getId(), initNetworkTopology(allAppNames));

        return clusterConfigs;
    }

    @Override
    public List<SeparatedClusterConfig> sendClusterConfigToEdgeNodeByThreads(List<SeparatedClusterConfig> clusterConfigs) {
        if(clusterConfigs.size() == 0) {
            LOG.warn("下发的集群配置为空。");
            return clusterConfigs;
        }
        List<String> allAppNames = new ArrayList<>();
        // 这里应该做一个判断，edgeNode的ip是否可用，如果不可用，或者未设定，将随机挑选一个进行设置
        // TODO:这里使用一个线程进行创建，加快docker启动速度

        clusterConfigs.forEach(c -> {

                    // 如果edgeId为空，随机挑选一个
                    if (c.getEdgeNodeId() == null || "".equals(c.getEdgeNodeId())) {
                        String edgeNode = chooseEdgeNodes(1).get(0);
                        c.setEdgeNodeId(edgeNode);
                    }
                    // 创建工作线程
                    InitEdgeThread initEdgeThread = new InitEdgeThread(restTemplate, c);
                    // 提交到工作线程中
                    initEdgeExcutorCompletionPool.submit(initEdgeThread);
                });
        // 获取返回值
        for (int i = 0; i < clusterConfigs.size(); i++) {
            try {
                Future<ThreadResult> result = initEdgeExcutorCompletionPool.take();
                ThreadResult threadResult = result.get();
                SeparatedClusterConfig c = threadResult.getSeparatedClusterConfig();

                if (threadResult.getResult() == null) {
                    continue;
                }

                // 更新配置
                //c.setClusterConfig(response.getBody());
                // TODO: 存储appName和containerId的映射
                if (!allContainerIds.containsKey(c.getClusterConfig().getId())) {
                    allContainerIds.put(c.getClusterConfig().getId(), new HashMap<>());
                }
                Map<String, Map<String, Set<String>>> allNames = allContainerIds.get(c.getClusterConfig().getId());
                if (!allNames.containsKey(c.getEdgeNodeId())) {
                    allNames.put(c.getEdgeNodeId(), new HashMap<>());
                }
                Map<String, Set<String>> edgeContainerIds = allNames.get(c.getEdgeNodeId());
                Map<String, List<String>> nameToIds = threadResult.getResult();
                nameToIds.forEach((name, v) -> {
                    if (!edgeContainerIds.containsKey(name)) {
                        edgeContainerIds.put(name, new HashSet<>());
                    }
                    edgeContainerIds.get(name).addAll(v);
                });
                // TODO: 对应需要设计一个删除这个配置的操作
                // 构建初始化的逻辑拓扑
                for (String appName : c.getClusterConfig().getTopology().getAppNames()) {
                    if (!appName.startsWith("br")) {
                        allAppNames.add(appName);
                        if (!allAppStatus.containsKey(c.getClusterConfig().getId())) {
                            allAppStatus.put(c.getClusterConfig().getId(), new HashMap<>());
                        }
                        // 保存appnames
                        Map<String, Set<String>> clusterAppConfig = allAppStatus.get(c.getClusterConfig().getId());
                        if (!clusterAppConfig.containsKey(appName)) {
                            clusterAppConfig.put(appName, new HashSet<>());
                        }
                        Set<String> edgeNodes = clusterAppConfig.get(appName);
                        edgeNodes.add(c.getEdgeNodeId());
                    }
                }
                if (!allClusterConfig.containsKey(c.getClusterConfig().getId())) {
                    allClusterConfig.put(c.getClusterConfig().getId(), new ArrayList<>());
                }
                allClusterConfig.get(c.getClusterConfig().getId()).add(c);
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        }
        allLogicalTopo.put(clusterConfigs.get(0).getClusterConfig().getId(), initNetworkTopology(allAppNames));

        return clusterConfigs;
    }

    private static class ThreadResult {
        private SeparatedClusterConfig separatedClusterConfig;
        private Map result;

        public ThreadResult(SeparatedClusterConfig separatedClusterConfig, Map result) {
            this.separatedClusterConfig = separatedClusterConfig;
            this.result = result;
        }

        public SeparatedClusterConfig getSeparatedClusterConfig() {
            return separatedClusterConfig;
        }

        public Map getResult() {
            return result;
        }
    }

    private static class InitEdgeThread implements Callable<ThreadResult> {
        private static final Logger LOG = LoggerFactory.getLogger(InitEdgeThread.class);
        private RestTemplate restTemplate;
        // 实际上为EdgeNode的Ip地址
        private SeparatedClusterConfig separatedClusterConfig;

        public InitEdgeThread(RestTemplate restTemplate, SeparatedClusterConfig separatedClusterConfig) {
            this.restTemplate = new RestTemplate();
            this.separatedClusterConfig = separatedClusterConfig;
        }

        @Override
        public ThreadResult call() throws Exception {
            LOG.info("边缘节点:{}正在初始化", separatedClusterConfig.getEdgeNodeId());
            ResponseEntity<Map> response = restTemplate.postForEntity("http://" + separatedClusterConfig.getEdgeNodeId()
                    + CREATE_CLUSTER, separatedClusterConfig.getClusterConfig(), Map.class);
            if(!response.getStatusCode().is2xxSuccessful()) {
                LOG.error("集群配置下发失败:{}:{}", separatedClusterConfig.getEdgeNodeId(), response.getBody());
                return null;
            }
            LOG.info("边缘节点:{}初始化完成", separatedClusterConfig.getEdgeNodeId());
            Map body = response.getBody();
            LOG.info("执行完成！");
            return new ThreadResult(separatedClusterConfig, body);
        }
    }

    private NetworkTopology initNetworkTopology(List<String> allAppNames) {
        NetworkTopology networkTopology = new NetworkTopology();
        networkTopology.setTopologyId(1);
        networkTopology.setAppNames(allAppNames.toArray(new String[allAppNames.size()]));
        int[][] topo = new int[allAppNames.size()][allAppNames.size()];
        for (int i = 0; i < allAppNames.size(); i++) {
            for (int j = 0; j < allAppNames.size(); j++) {
                topo[i][j] = 1;
            }
        }
        networkTopology.setTopology(topo);
        return networkTopology;
    }

    /**
     * 删除全部的容器集群
     * @param clusterId
     * @return
     */
    @Override
    public boolean deleteClusterFromEdgeNode(long clusterId) {
        Map<String, Object> parameters = new HashMap<>();
        parameters.put("clusterId", clusterId);
        if(!allClusterConfig.containsKey(clusterId)) {
            LOG.warn("集群{}不存在");
            return false;
        }

        List<SeparatedClusterConfig> separatedClusterConfigs = allClusterConfig.get(clusterId);
        separatedClusterConfigs.forEach(c -> {
            // 调用远程借口删除服务
            String res = restTemplate.getForObject("http://" + c.getEdgeNodeId() + DELETE_CLUSTER + "?clusterId=" + clusterId, String.class);
            LOG.info("删除集群{}", res);
        });
        // 删除对应的逻辑topo
        allLogicalTopo.remove(clusterId);
        return true;
    }

    /**
     * 获取cluster的全网拓扑
     * @param clusterId
     * @return
     */
    public NetworkTopology getClusterTopology(long clusterId) {
        return null;
    }

    /**
     * 设置网络的逻辑拓扑,
     * @param clusterId
     * @param topology 这里指的是逻辑拓扑
     */
    @Override
    public void updateLogicalTopology(long clusterId, NetworkTopology topology) {
        // 校验下topo是否合法
        for (String appName : topology.getAppNames()) {
            if(appName.contains("br")) {
                throw new RuntimeException("暂时不支持非逻辑网络拓扑！");
            }
        }
        // 更新topo配置
        // 获取原来的网络topo
        if (!allLogicalTopo.containsKey(clusterId)) {
            throw new RuntimeException("不存在的topo，无法进行更新");
        }

        NetworkTopology networkTopology = allLogicalTopo.get(clusterId);
        String[] appNames = networkTopology.getAppNames();
        if(appNames.length != topology.getAppNames().length) {
            throw new RuntimeException("更新的topo不一致");
        }
        for (int i = 0; i < appNames.length; i++) {
            // 这里的名称还需要在做处理
            if (!appNames[i].equals(topology.getAppNames()[i])) {
                throw new RuntimeException("更新的名称不一致");
            }
        }

        int[][] oldTopo = networkTopology.getTopology();
        int[][] newTopo = topology.getTopology();

        // 对比更新
        for (int i = 1; i < appNames.length; i++) {
            for (int j = 0; j < i; j++) {
                if(oldTopo[i][j] == 1 && newTopo[i][j] == 0) {
                    // 阻断连接
                    // 找到appNames对应的边缘服务器
                    Set<String> edgeNodesIdSrc = allAppStatus.get(clusterId).get(appNames[i]);
                    Set<String> edgeNodesIdDst = allAppStatus.get(clusterId).get(appNames[j]);
                    // 获取AppName对应的全部容器IP地址
                    Set<String> appName1List = getIpListFromRemoteIpServiceByAppName(clusterId, appNames[i]);
                    Set<String> appName2List = getIpListFromRemoteIpServiceByAppName(clusterId, appNames[j]);

                    String dstAppName = appNames[i];
                    edgeNodesIdSrc.forEach(edgeIp -> {
                        // 调用远程阻断连接端口
                        Map<String, Object> param = new HashMap<>();
                        param.put("clusterId", clusterId);
                        param.put("appName", dstAppName);
                        param.put("ipList", appName2List.toArray());
                        ResponseEntity<String> forEntity = restTemplate.getForEntity("http://" + edgeIp + DROP_DOCKER_NETWORK, String.class, param);
                        if (!forEntity.getStatusCode().is2xxSuccessful()){
                            LOG.error("下发逻辑拓扑修改出现错误");
                        }
                    });
                    String srcAppName = appNames[j];
                    edgeNodesIdDst.forEach(edgeIp -> {
                        // 调用远程阻断连接端口
                        Map<String, Object> param = new HashMap<>();
                        param.put("clusterId", clusterId);
                        param.put("appName", srcAppName);
                        param.put("ipList", appName1List.toArray());
                        ResponseEntity<String> forEntity = restTemplate.getForEntity("http://" + edgeIp + DROP_DOCKER_NETWORK, String.class, param);
                        if (!forEntity.getStatusCode().is2xxSuccessful()){
                            LOG.error("下发逻辑拓扑修改出现错误");
                        }
                    });
                    // step2 调用远程接口，添加阻断连接
                }
                else if(oldTopo[i][j] == 0 && newTopo[i][j] == 1) {
                    // 创建连接
                    // step1 找到appNames对应的边缘服务器
                    // step2 调用远程接口，删除阻断连接
                    Set<String> edgeNodesIdSrc = allAppStatus.get(clusterId).get(appNames[i]);
                    Set<String> edgeNodesIdDst = allAppStatus.get(clusterId).get(appNames[j]);
                    // 获取AppName对应的全部容器IP地址
                    Set<String> appName1List = getIpListFromRemoteIpServiceByAppName(clusterId, appNames[i]);
                    Set<String> appName2List = getIpListFromRemoteIpServiceByAppName(clusterId, appNames[j]);

                    String dstAppName = appNames[i];
                    edgeNodesIdSrc.forEach(edgeIp -> {
                        // 调用远程阻断连接端口
                        Map<String, Object> param = new HashMap<>();
                        param.put("clusterId", clusterId);
                        param.put("appName", dstAppName);
                        param.put("ipList", appName2List.toArray());
                        ResponseEntity<String> forEntity = restTemplate.getForEntity("http://" + edgeIp + CANCEL_DROP_DOCKER_NETWORK, String.class, param);
                        if (!forEntity.getStatusCode().is2xxSuccessful()){
                            LOG.error("下发逻辑拓扑修改出现错误");
                        }
                    });
                    String srcAppName = appNames[j];
                    edgeNodesIdDst.forEach(edgeIp -> {
                        // 调用远程阻断连接端口
                        Map<String, Object> param = new HashMap<>();
                        param.put("clusterId", clusterId);
                        param.put("appName", srcAppName);
                        param.put("ipList", appName1List.toArray());
                        ResponseEntity<String> forEntity = restTemplate.getForEntity("http://" + edgeIp + CANCEL_DROP_DOCKER_NETWORK, String.class, param);
                        if (!forEntity.getStatusCode().is2xxSuccessful()){
                            LOG.error("下发逻辑拓扑修改出现错误");
                        }
                    });
                }
            }
        }
        networkTopology.setTopology(newTopo);
        allLogicalTopo.put(clusterId, networkTopology);
    }

    @Override
    public NetworkTopology getLogicalNetworkByClusterId(Long clusterId) {
        if(!allLogicalTopo.containsKey(clusterId)) {
            LOG.error("无法找到对应的集群topo");
            return null;
        }
        return allLogicalTopo.get(clusterId);
    }

    @Override
    public String cancelNetworkMonitor(Long clusterId, String appName) {
        if(!allAppStatus.containsKey(clusterId)) {
            LOG.warn("正在取消一个不存在集群的网络监控");
            return "failed";
        }
        if (!allAppStatus.get(clusterId).containsKey(appName)) {
            LOG.warn("正在取消一个不存在的AppName的监控");
            return "failed";
        }
        Set<String> nodeIds = allAppStatus.get(clusterId).get(appName);

        // 遍历clusterAppName下发，容器监控模块
        nodeIds.forEach(c -> {
            if (!"".equals(c)) {
                // 下发启用monitor监控的指令
                Map<String, Object> param = new HashMap<>();
                param.put("clusterId", clusterId);
                param.put("appName", appName);
                ResponseEntity<Boolean> response
                        = restTemplate.getForEntity("http://" + c + CANCEL_NETWORK_MONITOR, Boolean.class, param);
                if(!response.getStatusCode().is2xxSuccessful()) {
                    LOG.warn("集群{}通信失败", c);
                }
                if(!response.getBody()) {
                    LOG.warn("集群{},appName{}指令下发失败", c, appName);
                }
            }
        });
        return "success";
    }

    @Override
    public String enableNetworkMonitor(Long clusterId, String appName) {
        if(!allAppStatus.containsKey(clusterId)) {
            LOG.warn("正在启用一个不存在集群的网络监控");
            return "failed";
        }
        if (!allAppStatus.get(clusterId).containsKey(appName)) {
            LOG.warn("正在启用一个不存在的AppName的监控");
            return "failed";
        }
        Set<String> nodeIds = allAppStatus.get(clusterId).get(appName);

        // 遍历clusterAppName下发，容器监控模块
        nodeIds.forEach(c -> {
            if (!"".equals(c)) {
                // 下发启用monitor监控的指令
                Map<String, Object> param = new HashMap<>();
                param.put("clusterId", clusterId);
                param.put("appName", appName);
                ResponseEntity<Boolean> response
                        = restTemplate.getForEntity("http://" + c + ENABLE_NETWORK_MONITOR, Boolean.class, param);
                if(!response.getStatusCode().is2xxSuccessful()) {
                    LOG.warn("集群{}通信失败", c);
                }
                if(!response.getBody()) {
                    LOG.warn("集群{},appName{}指令下发失败", c, appName);
                }
            }
        });
        return "success";
    }

    @Override
    public List<ContainerMonitorInfo> getContainerMonitorInfoByAppName(Long clusterId, String appName) {
        List<ContainerMonitorInfo> containerMonitorInfos = new ArrayList<>();

        if(!allClusterConfig.containsKey(clusterId)) {
            return containerMonitorInfos;
        }
        List<SeparatedClusterConfig> separatedClusterConfigs = allClusterConfig.get(clusterId);

        // TODO: 这里边没有需要的containerId
        separatedClusterConfigs.forEach( c -> {
            if(c.getClusterConfig().getTopology().containsAppName(appName)) {
                for (ContainerDescription docker : c.getClusterConfig().getDockers()) {
                    // 从找到对应容器id
                    Set<String> cids = allContainerIds.get(clusterId).get(c.getEdgeNodeId()).get(appName);
                    cids.forEach( cid -> {
                        ContainerMonitorInfo monitorInfo = new ContainerMonitorInfo();
                        monitorInfo.setAppName(appName);
                        monitorInfo.setEdgeNodeId(c.getEdgeNodeId());
                        monitorInfo.setClusterId(clusterId);
                        monitorInfo.setContainerId(cid);
                        containerMonitorInfos.add(monitorInfo);
                    });
                }
            }
        });
        return containerMonitorInfos;
    }

    /**
     * 获取全部的边缘节点地址
     * @return
     */
    @Override
    public List<String> getEdgeNodeIds() {
        return cloudRegisterCenter.getAvailableEdgeNodes();
    }
}
